use std::{thread, time::Duration};
use tokio::{self, runtime::Runtime, sync, sync::oneshot, time};

//#[tokio::main(worker_threads = 5)]
fn main() {
    //rt_example();
    //enter_example();
    //blocking_example();
    //abort_example();
    //join_example();
    
    //oneshot_channel_example()
    mpsc_channel_example()
}

fn now() -> String {
    chrono::Local::now().format("%F %T.%3f").to_string()
}

fn rt_example() {
    // 创建runtime, 等价于使用tokio::main注解
    //let rt = tokio::runtime::Runtime::new().unwrap();
    //let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(3)
        .enable_io()
        .enable_time()
        .build()
        .unwrap();

    // block_on会阻塞当前线程, 直到异步任务完成
    let _ = rt.block_on(async {
        println!("before sleep: {}", now());
        let task = tokio::time::sleep(tokio::time::Duration::from_secs(1));
        task.await; // 调用异步任务
        println!("after sleep: {}", now());
    });

    // 在runtime中调用
    rt.block_on(async {
        my_func();
    });

    rt.block_on(async {
        my_func2(&rt);
    });

    // thread的sleep, 关闭rt时会阻塞
    rt.spawn_blocking(|| {
        thread::sleep(Duration::from_secs(5));
        println!("blocking thread task over: {}", now());
    });
    rt.spawn(async {
        thread::sleep(Duration::from_secs(1));
        println!("worker thread task over: {}", now());
    });

    // 删除rt句柄, 只会立即关闭未被阻塞的worker thread
    // 对已经运行的blocking thread 以及 已经阻塞整个线程的worker thread仍然会执行
    drop(rt);
}

// 在runtime外部定义异步任务
// - 通过tokio::spawn
// - 通过Runtime.spawn
fn my_func() {
    println!("create an async task");
    tokio::spawn(async {
        println!("spawn an async task");
        tokio::time::sleep(Duration::from_secs(2)).await;
    });
}

fn my_func2(rt: &Runtime) {
    rt.spawn(async {
        tokio::time::sleep(Duration::from_secs(2)).await;
    });
}

fn enter_example() {
    let rt = tokio::runtime::Runtime::new().unwrap();

    // 进入runtime, 但不阻塞当前线程
    let guard = rt.enter();
    // 已经处于runtime上下文中
    my_func();
    // 退出上下文
    drop(guard);

    println!("before sleep on main thread: {}", now());
    std::thread::sleep(Duration::from_secs(5));
}

// blocking thread相当于thread::spawn
// 不同点: 在runtime内部, 可使用await等异步操作
fn blocking_example() {
    let rt = Runtime::new().unwrap();
    // 创建一个blocking thread，可立即执行(由操作系统调度系统决定何时执行)
    // 注意，不阻塞当前线程
    let task = rt.spawn_blocking(|| {
        println!("in task: {}", now());
        // 注意，是线程的睡眠，不是tokio的睡眠，因此会阻塞整个线程
        thread::sleep(std::time::Duration::from_secs(1))
    });

    // 小睡1毫秒，让上面的blocking thread先运行起来
    std::thread::sleep(std::time::Duration::from_millis(1));
    println!("not blocking: {}", now());

    // 可在runtime内等待blocking_thread的完成
    rt.block_on(async {
        task.await.unwrap();
        println!("after blocking task: {}", now());
    });
}

fn abort_example() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        let task = tokio::spawn(async {
            println!("start blocking task");
            time::sleep(time::Duration::from_secs(10)).await;
            println!("finish block task");
        });

        time::sleep(time::Duration::from_millis(1)).await;
        task.abort();
        let err = task.await.unwrap_err();
        println!("abort: {}", err.is_cancelled())
    })
}

async fn do_one() -> Result<(), &'static str> {
    println!("doing one: {}", now());
    time::sleep(time::Duration::from_secs(2)).await;
    println!("do one done: {}", now());
    Ok(())
}

async fn do_two() -> Result<(), &'static str> {
    println!("doing two: {}", now());
    time::sleep(time::Duration::from_secs(1)).await;
    println!("do two done: {}", now());
    Err("do two err")
}

fn join_example() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        // 等待两个任务完成
        /* tokio::join!(do_one(), do_two());
        println!("all done: {}", now()); */

        let result = tokio::try_join!(do_one(), do_two());
        match result {
            Ok((_, _)) => {
                println!("all done")
            }
            Err(err) => {
                // 有一个err就提前返回
                println!("processing failed: {}", err)
            }
        }
    })
}

fn oneshot_channel_example() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        let (tx, mut rx) = oneshot::channel::<i32>();

        let mut tick1 = tokio::time::interval(Duration::from_secs(2));
        tick1.tick().await;

        tokio::spawn(async move {
            tick1.tick().await;
            println!("send msg {}", now());
            match tx.send(1) {
                Ok(_) => {
                    println!("send success");
                }
                Err(e) => {
                    println!("send error:{}", e)
                }
            }
        });

        let mut tick2 = tokio::time::interval(Duration::from_secs(1));
        loop {
            tokio::select! {
                _ = tick2.tick() => println!("tick..."),
                msg = &mut rx => {
                    println!("Got message: {}", msg.unwrap());
                    break; // 接收后rx被消耗
                }
            }
        }
    });

    thread::sleep(Duration::from_secs(5));
}

fn mpsc_channel_example() {
    let rt = Runtime::new().unwrap();
    rt.block_on(async {
        let (tx, mut rx) = sync::mpsc::channel::<i32>(100);
        
        // spawn的异步任务活的更久, 需要获取tx的所有权 
        /*rt.spawn(async move {
            for i in 1..=10 {
                println!("send {}", i);
                if tx.send(i).await.is_err() {
                    println!("send err: {}", i);
                    break
                }
            }
        });*/

        // 每个异步任务都持有tx
        for i in 1..=10 {
            let tx = tx.clone();
            rt.spawn(async move {
                println!("send {}", i);
                if tx.send(i).await.is_err() {
                    println!("send err: {}", i);
                }
            });
        }
        drop(tx); // 需关闭原始任务中的tx, 否则rx不会返回None

        while let Some(i) = rx.recv().await {
            println!("receive: {}", i);
        }
        
    });
    
    thread::sleep(Duration::from_secs(5));
}
